package com.intct.func;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.intct.utils.HBaseUtil;
import com.intct.utils.LettuceUtil;
import io.lettuce.core.api.StatefulRedisConnection;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;
import org.apache.hadoop.hbase.client.AsyncConnection;

import java.util.Collections;
import java.util.concurrent.CompletableFuture;
import java.util.function.Function;
import java.util.function.Supplier;

/**
 * @author gufg
 * @since 2025-09-19 15:31
 */
public class LookupAsyncDimFun extends RichAsyncFunction<String, String> {

    private StatefulRedisConnection<String, String> redisAsyncConnection;
    private AsyncConnection asyncHBaseConn;

    private String rowkey;
    private String namespace;
    private String tableName;

    public LookupAsyncDimFun(String namespace, String tableName, String rowkey) {
        this.rowkey = rowkey;
        this.namespace = namespace;
        this.tableName = tableName;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        redisAsyncConnection = LettuceUtil.getRedisAsyncConnection();
        asyncHBaseConn = HBaseUtil.getAsyncConnection();
    }

    @Override
    public void close() throws Exception {
        LettuceUtil.closeRedisAsyncConnection(redisAsyncConnection);
        HBaseUtil.closeAsycnConnection(asyncHBaseConn);
    }

    @Override
    public void asyncInvoke(String jsonStr, ResultFuture<String> resultFuture) throws Exception {
        JSONObject inputObj = JSON.parseObject(jsonStr);
        // 设置客户端完成请求后要执行的回调函数
        // 回调函数只是简单地把结果发给 future
        CompletableFuture.supplyAsync(new Supplier<String>() {
            @Override
            public String get() {
                // Redis查询数据是否存在
                String rowkeyValue = inputObj.getString(rowkey);
                JSONObject readDimAsync = LettuceUtil.readDimAsync(redisAsyncConnection, rowkeyValue);
                if (readDimAsync == null) {
                    return null;
                }
                return LettuceUtil.readDimAsync(redisAsyncConnection, rowkeyValue).toJSONString();
            }
        }).thenApplyAsync(new Function<String, String>() {   // 有入参，
            @Override
            public String apply(String jsonStr) {
                // 判断Redis是否有数据，没数据时从HBase查询
                if (jsonStr == null) {
                    String rowkeyValue = inputObj.getString(rowkey);
                    JSONObject jsonObj = HBaseUtil.getRowkeyData(asyncHBaseConn, namespace, tableName, rowkeyValue);

                    // 如果HBase查找到数据
                    if (jsonObj != null && jsonObj.size() > 0) {
                        // 写到Redis中一份
                        LettuceUtil.writeDimAsync(redisAsyncConnection, rowkeyValue, jsonObj);
                    }

                    return jsonObj.toJSONString();
                }

                return jsonStr;
            }
        }).thenAccept((String dbJsonStrResult) -> {  // 返回数据
            // redis hbase其中一个查询数据
            if (dbJsonStrResult != null) {
                // 查询结果
                JSONObject queryObj = JSON.parseObject(dbJsonStrResult);
                // 将维度数据放入到主流数据
                inputObj.putAll(queryObj);
            }

            resultFuture.complete(Collections.singleton(JSON.toJSONString(inputObj)));
        });
    }
}